/**
 * Generic IPC operations.
 *
 * for mq_ see: http://www.helsinki.fi/atk/unix/dec_manuals/DOC_40D/APS33DTE/DOCU_011.HTM
 */

#include "ipcops.h"

#include <mqueue.h>
#include <errno.h>

#include <fcntl.h>           /* For O_* constants */
#include <sys/stat.h>        /* For mode constants */

#include <stdio.h>
#include <stdlib.h>

#include <string.h>

// For defining va_list etc.
#include <stdarg.h>

#include "FunctionTracing.h"

// TODO V Fix it so it returns the right RC value. And possibly change all callers to handle the RC correct.
/**
 * @param arFileDescriptor -- int array of 2. 0: receiving end. 1: transmitting end.
 */
int PipeCreate(int arFileDescriptor[2]) {
	int nRc = 0;
	// TRACEF(7,("PIPE_BUF=%d\n", PIPE_BUF));

	if (pipe(arFileDescriptor) < 0) {
		perror("pipe");
		exit(-1);
	}

	TRACEF(7,("OpenPipe(%d, %d)\n", arFileDescriptor[0], arFileDescriptor[1]));
	return (nRc);
} // openpipe


int PipeReceive(int nPipeHandle, void *pBuffer, int nLength) {
	int nRc = 0;

	nRc = read(nPipeHandle, pBuffer, nLength);

	return (nRc);
} // end pipereceive


/*
 * Example:
 * nRc = PipSend(arFileDescriptor[1], pBuffer, nBufLen);
 */
int PipeSend(int nPipeHandle, void *pBuffer, int nLength) {
	int nRc = 0;

	nRc = write(nPipeHandle, pBuffer, nLength);

	return (nRc);
} // end pipereceive

/**
 * Create a MQ message queue that does not exist yet.
 *
 *@param szQueueName -- Name of queue. Mandatory.
 *@param nMsgTextMaxSize -- Maximum size of the message text. Default to DATA_LOGGER_MSG_TEXT_SIZE.
 *@param nMaxMsgsInQueue -- Maximum number of messages in the queue. default to MAX_MQ_MSG_IN_QUE.
 *
 * @example
 mqd_t nSubAgentHiTimeQueId;

 mqd_t nMsgQueId = CreateMqMessage(g_szRtAgentReportQue, CMD_MESSAGE_SIZE, CMD_MSG_PER_QUEUE);
 if (nMsgQueId == -1) {
 // TODO N If failed report back to broker
 nRc = -1; // indicate fail so no further execution will take place.
 }
 */
mqd_t CreateMqMessage(const char *szQueueName, int nMsgTextMaxSize,
		int nMaxMsgsInQueue) {
	mqd_t nMsgId = -1; /* Message queue IPC ID. */

	struct mq_attr stMqAttr;

	stMqAttr.mq_maxmsg = nMaxMsgsInQueue; // Maximum number of messages in the queue.
	// Max. message size (bytes)
	stMqAttr.mq_msgsize = nMsgTextMaxSize;
	stMqAttr.mq_flags = 0; // No special flags.

	// TODO N Re-write the code so that this section will only be done if the szQueueName exists.
	DestroyMqMessage(szQueueName);

	TRACEF(7,("CreateMqMessage() Creating \"%s\" with msgs(%d) of size(%d)\n",
			szQueueName, nMaxMsgsInQueue, nMsgTextMaxSize));
	/*
	 * O_CREAT - Create the queue if it doesn't exits.
	 * O_EXCL  - Exclusive - Fail the operation if the queue already exists.
	 * O_RDRW  - read and write access.
	 */
	nMsgId = mq_open(szQueueName, O_CREAT | O_EXCL | O_RDWR, 0700, &stMqAttr);
	if (nMsgId == -1) {
		perror("mq_open(create)");
		exit(-1);
	} // endif msggid == -1

	return (nMsgId);
} // end CreateMqMessage.



/**
 * Removes the szQueueName if it exists.
 *
 * @param
 * @return -- 0 if ok, otherwise -1.
 */
int DestroyMqMessage(const char *szQueueName) {
	int nRc = -1;

	// TODO V exit if szQueueName is null.
	// Create the Message queue to use for data logging.
	// See Linux programming in 24 hours, p415.
	nRc = mq_unlink(szQueueName);
	if (nRc == -1) {
		perror("mq_unlink()");
		//exit(-1);
	} // endif nrc	 == -1

	return (nRc);
} // end destroy mq message.




/**
 * Open an MQ message queue that already exists.
 *
 */
mqd_t OpenMqMessage(const char *szQueueName) {
	mqd_t nMsgId = -1; /* Message queue IPC ID. */

	nMsgId = mq_open(szQueueName, O_RDWR);
	if (nMsgId == -1) {
		TRACEF(5,("OpenMqMessage('%s') failed\n", szQueueName));
		perror("OpenMqMessage(): mq_open()");
		exit(-1);
	} // endif msggid == -1


	return (nMsgId);
}

/**
 * @param szMessage -- pointer to a pre-allocated char array.
 * @param nMaxLength -- the allocation size of szMessage.
 * @param nRecoverOnIntr -- if set, then retry read if EINTR caused the receive break.
 *
 * @return nRc -- Returns the length of the message, or -1 on error, -2 if read was interrupted with EINTR.
 */
int ReceiveMqMessage(mqd_t nQueId, char *szMessage, int nMaxLength, int nRecoverOnIntr) {
	int nRc = 0;
	unsigned nPriority = 0; // 0 means all.

	do {
		// TODO C do this the same way as in the PipeReceive instance.
		nRc = mq_receive(nQueId, szMessage, nMaxLength, &nPriority);
	} while (nRc == -1 && errno == EINTR && nRecoverOnIntr);
	if (nRc == -1) {
		if (errno == EINTR) {
			TRACEF(7,("ReceiveMqMessage(): mq_receive() interrupted.\n"));
			nRc = -2;
		} else {
			TRACEF(6,("ReceiveMqMessage(%u)\n", nQueId));
			perror("mq_receive()");
		}
	} else {// endif msggid == -1
		TRACEF(7,("ReceiveMqMessage(%d) recieved: '%s' (%d of %d)\n", nRc,
				szMessage, (int) strlen(szMessage), nMaxLength));
	}
	return (nRc);
} // end receivemqmessage.


/**
 * @note This function will block if the message queue is full.
 */
int SendMqMessage(int nMsgId, char *szMessage, int nMsgLength) {
	int nRc = 0;
	unsigned nPriority = 0; // 0 is the lowest priority.

	do {
		nRc = mq_send(nMsgId, szMessage, nMsgLength, nPriority);
	} while (nRc == -1 && errno == EINTR);
	if (nRc == -1) {
		perror("mq_send()");
		exit(-1);
		//abort();
	} // endif msggid == -1
	return (nRc);
} // end send mq message.
